package com.leilei;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author lei
 * @version 1.0
 * @date 2021/4/10 21:32
 * @desc flatmap 算子
 */
public class FlatMapOperator {
    public static void main(String[] args) throws Exception {
        List<String> str1 = Arrays.asList("a", "b", "c");
        List<String> str2 = Arrays.asList("关羽", "张飞", "马超", "黄忠", "赵云");
        List<List<String>> originalData = new ArrayList<>();
        originalData.add(str1);
        originalData.add(str2);
        List<String> flatMapChangeList = originalData.stream().flatMap(Collection::stream).collect(Collectors.toList());
        // [a, b, c, 关羽, 张飞, 马超, 黄忠, 赵云]
        System.out.println(flatMapChangeList);


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度为1 （1个线程执行，以便观察）
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        // 加载数据源
        DataStreamSource<List<String>> stream = env.fromCollection(originalData);
        // 使用flatMap算子
        SingleOutputStreamOperator<String> source = stream.flatMap(new FlatMapFunction<List<String>, String>() {
            @Override
            public void flatMap(List<String> value, Collector<String> out) throws Exception {
                for (String s : value) {
                    out.collect(s);
                }
            }
        });
        // a b c 关羽 张飞 马超 黄忠 赵云
        source.print();
        env.execute();
    }
}
